package drds.global_transaction.transaction_coordinator.transaction_coordinator;

import drds.common.$;
import drds.common.Constants;
import drds.global_transaction.api.api.BranchTransactionStatus;
import drds.global_transaction.api.api.GlobalTransactionStatus;
import drds.global_transaction.api.api.ResourceManagerInbound;
import drds.global_transaction.api.api.RpcContext;
import drds.global_transaction.api.exception.GlobalTransactionException;
import drds.global_transaction.api.in_bound_handler.TransactionMessageHandler;
import drds.global_transaction.api.message_codec.RequestMessage;
import drds.global_transaction.api.message_codec.ResponseMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.TransactionResponse;
import drds.global_transaction.api.message_codec.request_and_response.transaction.branch_transaction.branch_transaction_end_status_report.branch_transaction_commit_status_report.BranchTransactionCommitRequestMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.branch_transaction.branch_transaction_end_status_report.branch_transaction_commit_status_report.BranchTransactionCommitResponseMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.branch_transaction.branch_transaction_end_status_report.branch_transaction_rollback_status_report.BranchTransactionRollbackRequestMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.branch_transaction.branch_transaction_end_status_report.branch_transaction_rollback_status_report.BranchTransactionRollbackResponseMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.GlobalTransactionRequestMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.branch_transaction_register.BranchTransactionRegisterRequestMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.branch_transaction_register.BranchTransactionRegisterResponseMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.branch_transaction_status_report.BranchTransactionStatusReportRequestMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.branch_transaction_status_report.BranchTransactionStatusReportResponseMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.global_transaction_begin.GlobalTransactionBeginRequestMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.global_transaction_begin.GlobalTransactionBeginResponseMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.global_transaction_end_status_report.GlobalTransactionStatusReportRequestMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.global_transaction_end_status_report.GlobalTransactionStatusReportResponseMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.global_transaction_end_status_report.global_transaction_commit_status_report.GlobalTransactionCommitRequestMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.global_transaction_end_status_report.global_transaction_commit_status_report.GlobalTransactionCommitResponseMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.global_transaction_end_status_report.global_transaction_rollback_status_report.GlobalTransactionRollbackRequestMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.global_transaction_end_status_report.global_transaction_rollback_status_report.GlobalTransactionRollbackResponseMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.global_transaction_lock_record_ids_check.GlobalTransactionLockRecordIdsCheckRequestMessage;
import drds.global_transaction.api.message_codec.request_and_response.transaction.global_transaction.global_transaction_lock_record_ids_check.GlobalTransactionLockRecordIdsCheckResponseMessage;
import drds.global_transaction.api.rpc.rpc_server.MessageSend;
import drds.global_transaction.common.GlobalTransactionIds;
import drds.global_transaction.common.thread.NamedThreadFactory;
import drds.global_transaction.transaction_coordinator.AbstractTransactionCoordinatorInboundHandler;
import drds.global_transaction.transaction_coordinator.transaction_session.BranchTransactionSession;
import drds.global_transaction.transaction_coordinator.transaction_session.GlobalTransactionSession;
import drds.global_transaction.transaction_coordinator.transaction_session.TransactionSessionManagerFactory;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.Collection;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import static drds.global_transaction.api.exception.TransactionExceptionCode.failed_send_branch_transaction_commit_request;
import static drds.global_transaction.api.exception.TransactionExceptionCode.failed_send_branch_transaction_rollback_request;

@Slf4j
public class TransactionCoordinator extends AbstractTransactionCoordinatorInboundHandler implements TransactionMessageHandler, ResourceManagerInbound
{
	private MessageSend messageSend;
	private GlobalTransactionManagers globalTransactionManagers = GlobalTransactionManagersFactory.get();
	//
	private ScheduledThreadPoolExecutor retryCommitting = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("RetryCommitting", 1));
	private ScheduledThreadPoolExecutor asyncCommitting = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("AsyncCommitting", 1));
	private ScheduledThreadPoolExecutor retryRollbacking = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("RetryRollbacking", 1));
	private ScheduledThreadPoolExecutor timeoutCheck = new ScheduledThreadPoolExecutor(1, new NamedThreadFactory("TxTimeoutCheck", 1));

	public TransactionCoordinator(MessageSend messageSend)
	{
		this.messageSend = messageSend;
		globalTransactionManagers.setResourceManagerInbound(this);
	}
//

	public void init()
	{
		retryCommitting.scheduleAtFixedRate(new Runnable()
		{
			@Override
			public void run()
			{
				try
				{
					handleRetryCommitting();
				} catch (Exception e)
				{
					log.info("Exception retry committing ... ", e);
				}

			}
		}, 0, 5, TimeUnit.MILLISECONDS);

		asyncCommitting.scheduleAtFixedRate(new Runnable()
		{
			@Override
			public void run()
			{
				try
				{
					handleAsyncCommitting();
				} catch (Exception e)
				{
					log.info("Exception async committing ... ", e);
				}

			}
		}, 0, 10, TimeUnit.MILLISECONDS);
		//
		retryRollbacking.scheduleAtFixedRate(new Runnable()
		{

			@Override
			public void run()
			{
				try
				{
					handleRetryRollbacking();
				} catch (Exception e)
				{
					log.info("Exception retry rollbacking ... ", e);
				}

			}
		}, 0, 5, TimeUnit.MILLISECONDS);

		timeoutCheck.scheduleAtFixedRate(new Runnable()
		{
			@Override
			public void run()
			{
				try
				{
					timeoutCheck();
				} catch (Exception e)
				{
					log.info("Exception timeout checking ... ", e);
				}

			}
		}, 0, 2, TimeUnit.MILLISECONDS);

	}

	private void handleRetryCommitting()
	{
		Collection<GlobalTransactionSession> globalTransactionSessionList = TransactionSessionManagerFactory.getRetryCommitTransactionSessionManager().getGlobalTransactionSessionList();
		for (GlobalTransactionSession globalTransactionSession : globalTransactionSessionList)
		{
			try
			{
				globalTransactionManagers.doGlobalTransactionCommit(globalTransactionSession, true);
			} catch (GlobalTransactionException ex)
			{
				log.info("failed to retry committing [" + globalTransactionSession.getGlobalTransactionId() + "] " + ex.getTransactionExceptionCode() + " " + ex.getMessage());
			}
		}
	}

	private void handleAsyncCommitting() throws GlobalTransactionException
	{
		Collection<GlobalTransactionSession> globalTransactionSessionList = TransactionSessionManagerFactory.getAsynchronousTransactionSessionManager().getGlobalTransactionSessionList();
		for (GlobalTransactionSession globalTransactionSession : globalTransactionSessionList)
		{
			try
			{
				globalTransactionManagers.doGlobalTransactionCommit(globalTransactionSession, true);
			} catch (GlobalTransactionException ex)
			{
				log.info("failed to async committing [" + globalTransactionSession.getGlobalTransactionId() + "] " + ex.getTransactionExceptionCode() + " " + ex.getMessage());
			}
		}
	}

	private void handleRetryRollbacking()
	{
		Collection<GlobalTransactionSession> globalTransactionSessionList = TransactionSessionManagerFactory.getRetryRollbackTransactionSessionManager().getGlobalTransactionSessionList();
		for (GlobalTransactionSession globalTransactionSession : globalTransactionSessionList)
		{
			try
			{
				globalTransactionManagers.doGlobalTransactionRollback(globalTransactionSession, true);
			} catch (GlobalTransactionException ex)
			{
				log.info("failed to retry rollbacking [" + globalTransactionSession.getGlobalTransactionId() + "] " + ex.getTransactionExceptionCode() + " " + ex.getMessage());
			}
		}
	}

	private void timeoutCheck() throws GlobalTransactionException
	{
		Collection<GlobalTransactionSession> globalTransactionSessionList = TransactionSessionManagerFactory.getRootTransactionSessionManager().getGlobalTransactionSessionList();
		if (!$.isHasData(globalTransactionSessionList))
		{
			return;
		}
		log.info(GlobalTransactionSession.class.getSimpleName() + " timeout check begin");
		//
		for (GlobalTransactionSession globalTransactionSession : globalTransactionSessionList)
		{
			if (Constants.developMode)
			{
				log.debug(globalTransactionSession.getGlobalTransactionId() + " " + globalTransactionSession.getGlobalTransactionStatus() + " " + globalTransactionSession.getBeginDataTimeLong() + " " + globalTransactionSession.getTimeout());
			}
			//
			if (globalTransactionSession.getGlobalTransactionStatus() != GlobalTransactionStatus.begin || !globalTransactionSession.isTimeout())
			{
				continue;
			}
			log.warn("Global transaction[" + globalTransactionSession.getGlobalTransactionId() + "] is timeout and " + "will be rolled back.");
			globalTransactionSession.close();
			globalTransactionSession.changeGlobalTransactionStatus(GlobalTransactionStatus.timeout_rollbacking);
			//
			globalTransactionSession.addSessionLifecycleListener(TransactionSessionManagerFactory.getRetryRollbackTransactionSessionManager());
			TransactionSessionManagerFactory.getRetryRollbackTransactionSessionManager().addGlobalTransactionSession(globalTransactionSession);

		}
		log.info(GlobalTransactionSession.class.getSimpleName() + " timeout check end");
	}

	//
	@Override
	public ResponseMessage onRequest(RpcContext rpcContext, RequestMessage request)
	{
		if (!(request instanceof GlobalTransactionRequestMessage))
		{
			throw new IllegalArgumentException();
		}
		GlobalTransactionRequestMessage transactionRequest = (GlobalTransactionRequestMessage) request;
		transactionRequest.setTCInboundHandler(this);

		return transactionRequest.handle(rpcContext);
	}

	@Override
	public void onResponse(RpcContext rpcContext, ResponseMessage response)
	{
		if (!(response instanceof TransactionResponse))
		{
			throw new IllegalArgumentException();
		}

	}

	//
	@Override
	protected void doGlobalTransactionBegin(//
			RpcContext rpcContext, //
			GlobalTransactionBeginRequestMessage globalTransactionBeginRequestMessage, //
			GlobalTransactionBeginResponseMessage globalTransactionBeginResponseMessage) throws GlobalTransactionException
	{
		globalTransactionBeginResponseMessage.setGlobalTransactionIdWithAddressAndPort(globalTransactionManagers.globalTransactionBegin(rpcContext.getApplicationId(), rpcContext.getTransactionServiceGroup(), globalTransactionBeginRequestMessage.getTransactionName(), globalTransactionBeginRequestMessage.getGlobalTransactionTimeout()));
	}

	@Override
	protected void doGlobalTransactionCommit(//
			RpcContext rpcContext, //
			GlobalTransactionCommitRequestMessage globalTransactionCommitRequestMessage, //
			GlobalTransactionCommitResponseMessage globalTransactionCommitResponseMessage) throws GlobalTransactionException
	{
		globalTransactionCommitResponseMessage.setGlobalTransactionStatus(globalTransactionManagers.globalTransactionCommit(GlobalTransactionIds.generateGlobalTransactionIdWithAddressAndPort(globalTransactionCommitRequestMessage.getGlobalTransactionId())));

	}

	@Override
	protected void doGlobalTransactionRollback(//
			RpcContext rpcContext, //
			GlobalTransactionRollbackRequestMessage globalTransactionRollbackRequestMessage, //
			GlobalTransactionRollbackResponseMessage globalTransactionRollbackResponseMessage) throws GlobalTransactionException
	{
		globalTransactionRollbackResponseMessage.setGlobalTransactionStatus(globalTransactionManagers.globalTransactionRollback(GlobalTransactionIds.generateGlobalTransactionIdWithAddressAndPort(globalTransactionRollbackRequestMessage.getGlobalTransactionId())));
	}

	@Override
	protected void doGlobalTransactionLockQuery(//
			RpcContext rpcContext, //
			GlobalTransactionLockRecordIdsCheckRequestMessage globalTransactionLockRecordIdsCheckRequestMessage, //
			GlobalTransactionLockRecordIdsCheckResponseMessage globalTransactionLockRecordIdsCheckResponseMessage) throws GlobalTransactionException
	{
		globalTransactionLockRecordIdsCheckResponseMessage.setLock(globalTransactionManagers.lockRecordIdsCheck(GlobalTransactionIds.generateGlobalTransactionIdWithAddressAndPort(globalTransactionLockRecordIdsCheckRequestMessage.getGlobalTransactionId()), globalTransactionLockRecordIdsCheckRequestMessage.getResourceId(), globalTransactionLockRecordIdsCheckRequestMessage.getTableNameAndLockRecordIdsStringsString()));
	}

	@Override
	protected void doGlobalTransactionStatus(//
			RpcContext rpcContext, //
			GlobalTransactionStatusReportRequestMessage globalTransactionStatusReportRequestMessage, //
			GlobalTransactionStatusReportResponseMessage globalTransactionStatusReportResponseMessage) throws GlobalTransactionException
	{
		globalTransactionStatusReportResponseMessage.setGlobalTransactionStatus(globalTransactionManagers.getGlobalTransactionStatus(GlobalTransactionIds.generateGlobalTransactionIdWithAddressAndPort(globalTransactionStatusReportRequestMessage.getGlobalTransactionId())));
	}

	//
	@Override
	protected void doBranchTransactionRegister(//
			RpcContext rpcContext, //
			BranchTransactionRegisterRequestMessage branchTransactionRegisterRequestMessage, //
			BranchTransactionRegisterResponseMessage branchTransactionRegisterResponseMessage) throws GlobalTransactionException
	{
		branchTransactionRegisterResponseMessage.setGlobalTransactionId(branchTransactionRegisterRequestMessage.getGlobalTransactionId());
		branchTransactionRegisterResponseMessage.setBranchTransactionId(globalTransactionManagers.branchTransactionRegister(branchTransactionRegisterRequestMessage.getResourceId(), rpcContext.getClientIp(), GlobalTransactionIds.generateGlobalTransactionIdWithAddressAndPort(branchTransactionRegisterRequestMessage.getGlobalTransactionId()), branchTransactionRegisterRequestMessage.getTableNameAndLockRecordIdsStringsString()));

	}

	@Override
	public BranchTransactionStatus branchTransactionCommit(//
			String globalTransactionIdWithAddressAndPort, //
			long branchTransactionId, //
			String resourceId, //
			String applicationData//
	) throws GlobalTransactionException
	{
		try
		{
			BranchTransactionCommitRequestMessage branchTransactionCommitRequestMessage = new BranchTransactionCommitRequestMessage();
			branchTransactionCommitRequestMessage.setGlobalTransactionIdWithAddressAndPort(globalTransactionIdWithAddressAndPort);
			branchTransactionCommitRequestMessage.setBranchTransactionId(branchTransactionId);
			branchTransactionCommitRequestMessage.setResourceId(resourceId);
			branchTransactionCommitRequestMessage.setApplicationData(applicationData);

			GlobalTransactionSession globalTransactionSession = TransactionSessionManagerFactory.findGlobalTransactionSession(GlobalTransactionIds.getGlobalTransactionId(globalTransactionIdWithAddressAndPort));
			BranchTransactionSession branchTransactionSession = globalTransactionSession.getBranchTransactionSession(branchTransactionId);

			BranchTransactionCommitResponseMessage branchTransactionCommitResponseMessage = (BranchTransactionCommitResponseMessage) messageSend.asynchronousSendRequestWithTimeoutGetResponse(branchTransactionSession.getClientIp(), globalTransactionSession.getApplicationId(), resourceId, branchTransactionCommitRequestMessage);
			return branchTransactionCommitResponseMessage.getBranchTransactionStatus();
		} catch (IOException e)
		{
			throw new GlobalTransactionException(failed_send_branch_transaction_commit_request, branchTransactionId + "/" + globalTransactionIdWithAddressAndPort, e);
		} catch (TimeoutException e)
		{
			throw new GlobalTransactionException(failed_send_branch_transaction_commit_request, branchTransactionId + "/" + globalTransactionIdWithAddressAndPort, e);
		}
	}

	@Override
	public BranchTransactionStatus branchTransactionRollback(String globalTransactionIdWithAddressAndPort, long branchTransactionId, String resourceId, String applicationData) throws GlobalTransactionException
	{
		try
		{
			BranchTransactionRollbackRequestMessage branchTransactionRollbackRequestMessage = new BranchTransactionRollbackRequestMessage();
			branchTransactionRollbackRequestMessage.setGlobalTransactionIdWithAddressAndPort(globalTransactionIdWithAddressAndPort);
			branchTransactionRollbackRequestMessage.setBranchTransactionId(branchTransactionId);
			branchTransactionRollbackRequestMessage.setResourceId(resourceId);
			branchTransactionRollbackRequestMessage.setApplicationData(applicationData);

			GlobalTransactionSession globalTransactionSession = TransactionSessionManagerFactory.findGlobalTransactionSession(GlobalTransactionIds.getGlobalTransactionId(globalTransactionIdWithAddressAndPort));
			BranchTransactionSession branchTransactionSession = globalTransactionSession.getBranchTransactionSession(branchTransactionId);

			BranchTransactionRollbackResponseMessage branchTransactionRollbackResponseMessage = (BranchTransactionRollbackResponseMessage) messageSend.asynchronousSendRequestWithTimeoutGetResponse(branchTransactionSession.getClientIp(), globalTransactionSession.getApplicationId(), resourceId, branchTransactionRollbackRequestMessage);
			return branchTransactionRollbackResponseMessage.getBranchTransactionStatus();
		} catch (IOException e)
		{
			throw new GlobalTransactionException(failed_send_branch_transaction_rollback_request, branchTransactionId + "/" + globalTransactionIdWithAddressAndPort, e);
		} catch (TimeoutException e)
		{
			throw new GlobalTransactionException(failed_send_branch_transaction_rollback_request, branchTransactionId + "/" + globalTransactionIdWithAddressAndPort, e);
		}
	}

	@Override
	protected void doBranchTransactionReport(//
			RpcContext rpcContext, //
			BranchTransactionStatusReportRequestMessage branchTransactionStatusReportRequestMessage, //
			BranchTransactionStatusReportResponseMessage branchTransactionStatusReportResponseMessage) throws GlobalTransactionException
	{
		globalTransactionManagers.branchTransactionReport(GlobalTransactionIds.generateGlobalTransactionIdWithAddressAndPort(branchTransactionStatusReportRequestMessage.getGlobaltransactionId()), branchTransactionStatusReportRequestMessage.getBranchTransactionId(), branchTransactionStatusReportRequestMessage.getBranchTransactionStatus(), branchTransactionStatusReportRequestMessage.getApplicationData());

	}
	//

}
